package org.huel.backend.websocket;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.huel.backend.config.WebSocketServer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Component
public class LAndAByName {

    private KafkaConsumer<String, String> kfkConsumer;

    @PostConstruct
    public void init() {
        // 消费者配置信息
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "likai:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "LAndAByName");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 构建 Kafka 消费者
        kfkConsumer = new KafkaConsumer<>(properties);
        kfkConsumer.subscribe(Collections.singletonList("LAndAByName1"));

        // 启动消费者线程
        new Thread(this::consumeMessages).start();
    }

    private void consumeMessages() {
        try {
            while (true) {
                // 拉取消息
                ConsumerRecords<String, String> conRecord = kfkConsumer.poll(Duration.ofMillis(2000));

                // 按班级分类的数据结构
                Map<String, JSONArray> classGroupedData = new HashMap<>();

                // 遍历 Kafka 消息
                for (ConsumerRecord<String, String> record : conRecord) {
                    // 将消息转换为 JSON 对象
                    JSONObject messageJson = JSONObject.parseObject(record.value());

                    // 提取字段
                    String classId = messageJson.getString("classId");
                    String studentName = messageJson.getString("studentName");
                    int presentCount = messageJson.getIntValue("L");
                    int absentCount = messageJson.getIntValue("A");

                    // 构造学生 JSON 对象
                    JSONObject studentJson = new JSONObject();
                    studentJson.put("studentName", studentName);
                    studentJson.put("L", presentCount);
                    studentJson.put("A", absentCount);

                    // 将学生数据添加到对应班级的分组中
                    classGroupedData
                            .computeIfAbsent(classId, k -> new JSONArray())
                            .add(studentJson);
                }

                // 生成输出 JSON
                JSONObject outputJson = new JSONObject();
                outputJson.putAll(classGroupedData);

                // 发送到 WebSocket
                WebSocketServer.sendMessage("LAndAByName", outputJson.toJSONString());
            }
        } catch (Exception e) {
            e.printStackTrace(); // 处理异常并输出错误信息
        } finally {
            // 确保消费者在结束时关闭
            kfkConsumer.close();
        }
    }
}
